package com.parsechina.hpush.broker.internal.listener;

import com.parsechina.hpush.broker.internal.message.NotifyPublishMessage;
import com.parsechina.hpush.broker.parser.PublishMessage;
import com.parsechina.hpush.broker.spi.ProtocolProcessor;
import io.netty.buffer.Unpooled;
import io.netty.handler.codec.mqtt.MqttQoS;

/**
 * @author linfeng
 * @date 2019-08-19
 **/
public class NotifyPublishMessageListener extends BaseMessageListener<NotifyPublishMessage> {


    NotifyPublishMessageListener(ProtocolProcessor protocolProcessor, String instanceId) {
        super(protocolProcessor, instanceId);
    }

    @Override
    public void onMessage(CharSequence channel, NotifyPublishMessage message) {

        if (!getInstanceId().equals(message.getUuid())
                && getProtocolProcessor().isExistClientOfServer(message.getClientId())) {

            PublishMessage publishMessage = new PublishMessage();
            publishMessage.setTopicName(message.getTopic());
            publishMessage.setQos(MqttQoS.valueOf(message.getQos()));
            publishMessage.setPayload(Unpooled.wrappedBuffer(message.getPayload()));
            publishMessage.setLocal(false);
            publishMessage.setRetained(message.isRetained());
            publishMessage.setClientId(message.getClientId());

            getProtocolProcessor().internalPublish(publishMessage);

        }


    }

}
